package com.ipinyou.mprofile

import kafka.serializer.StringDecoder
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext._
import scala.collection.mutable.HashMap

import java.text.SimpleDateFormat
import java.util.Calendar
import java.util.Date

object KafkaDataInput {

  val MILLIS_INDAY = 86400000

  val Z = "\u0002Z\u0002"

  val STRATEGY_ID = Set("137391", "171904", "171905")

  def isNumber(s: String): Boolean = s.matches("[+-]?\\d+\\.?\\d+")

  def main(args: Array[String]) {

    val sparkConf = new SparkConf().setAppName("AllDataInput")
    val ssc = new StreamingContext(sparkConf, Minutes(10))

    // checkpoint ?
    ssc.checkpoint("checkpoint")

    val brokers = "192.168.145.201:9092," +
      "192.168.145.202:9092," +
      "192.168.145.203:9092," +
      "192.168.145.204:9092," +
      "192.168.145.205:9092," +
      "192.168.145.206:9092," +
      "192.168.145.207:9092," +
      "192.168.145.208:9092," +
      "192.168.145.209:9092," +
      "192.168.145.210:9092"

    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)

    val setImpClk = Set("ic")
    val setBid = Set("bid")

    val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, setImpClk)

    //val bid_messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, setBid)

    val imp = messages.map(_._2)
      .map { line =>
        val p = line.split("\t")
        if (p.length > 95 && p(2).equalsIgnoreCase("1")) {
          val (actionId, actionFirstId, actionRequestTime, strategyId, bidPolicyData, bidPrice,
            winPrice, advertiserSrvPrice, platformOffsetPrice, agencySrvPrice) = (p(0), p(4), p(6), p(53), p(85), p(90), p(91), p(93), p(94), p(95));
          val other_cost = advertiserSrvPrice.toDouble + platformOffsetPrice.toDouble + agencySrvPrice.toDouble
          (List(actionFirstId, actionRequestTime, bidPrice, winPrice, other_cost.toString, strategyId, bidPolicyData).mkString("\t"));
        } else
          (Z)
      }.filter(!_.equalsIgnoreCase(Z))

    val click = messages.map(_._2)
      .map { line =>
        val p = line.split("\t")
        if (p.length > 85 && p(2).equalsIgnoreCase("2")) {
          val (actionId, actionFirstId, actionRequestTime, strategyId, bidPolicyData) = (p(0), p(4), p(6), p(53), p(85));
          (List(actionFirstId, actionRequestTime, strategyId, bidPolicyData).mkString("\t"));
        } else
          (Z)
      }.filter(!_.equalsIgnoreCase(Z))

    // 10 minutes?
    click.saveAsTextFiles("cpc_online/click_all/KafkaMultiStrategy", "10minutes")
    imp.saveAsTextFiles("cpc_online/imp_all/KafkaMultiStrategy", "10minutes")
    //bid.saveAsTextFiles("cpc_online/bid_all/KafkaMultiStrategy", "10minutes")
    
    
    ssc.start()
    ssc.awaitTermination()
  }
}
